package com.atguigu.bigdata.gmall.realtime.app.dwd;

import com.atguigu.bigdata.gmall.realtime.app.BaseSQLApp;
import com.atguigu.bigdata.gmall.realtime.common.Constant;
import com.atguigu.bigdata.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Dwd_04_ExamDetail extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_04_ExamDetail().init(
                4001,
                2,
                "Dwd_04_ExamDetail"
        );
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        //读取ods_db数据
        readOdsDb(tEnv,"Dwd_04_ExamDetail");

        //tEnv.sqlQuery("select * from ods_db where `table` = 'test_paper'").execute().print();
        //从ods中过滤出test_exam
        Table ExamInfo = tEnv.sqlQuery("select " +
                " `data`['id'] id, " +
                " `data`['paper_id'] paper_id, " +
                " `data`['user_id'] user_id, " +
                " `data`['score'] score, " +
                " `data`['duration_sec'] duration_sec, " +
                " `data`['create_time'] create_time, " +
                " ts," +
                "pt  " +
                "from ods_db " +
                "where `database`='gmall2022' " +
                "and `table`='test_exam' " +
                "and `type`='insert' " );
        tEnv.createTemporaryView("Exam_Info", ExamInfo);
        //ExamInfo.execute().print();

        //从ods中过露出test_paper
        Table paperInfo = tEnv.sqlQuery("select " +
                " `data`['id'] id, " +
                " `data`['paper_title'] paper_title, " +
                " `data`['course_id'] course_id, " +
                " `data`['create_time'] create_time, " +
                " ts " +
                "from ods_db " +
                "where `database`='gmall2022' " +
                "and `table`='test_paper' " +
                "and `type`='insert' " );
        tEnv.createTemporaryView("test_paper", paperInfo);
        //paperInfo.execute().print();

        //两张表join
        Table result = tEnv.sqlQuery("select " +
                "ei.id,  " +
                "paper_title,  " +
                "paper_id, " +
                "course_id, " +
                "score,  " +
                "duration_sec,  " +
                "ei.create_time,  "+
                "ei.ts  " +
                "from Exam_Info ei " +
                "join test_paper  dic " +
                "on ei.paper_id=dic.id");
        //result.execute().print();打印会阻塞
        //写出到kafka中
        tEnv.executeSql("create table dwd_exam_detail(" +
                "id string,  " +
                "paper_title string,  " +
                "paper_id string,  " +
                "course_id string, " +
                "score string,  " +
                "duration_sec string,  " +
                "create_time string,  " +
                "ts bigint  " +
                ")" + SQLUtil.getKafkaSink(Constant.TOPIC_DWD_EXAM_DETAIL));

        result.executeInsert("dwd_exam_detail");


    }
}

